package com.hzh.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}


object Demo4MysqlSource {
  def main(args: Array[String]): Unit = {
    /**
     * 1、创建环境
     */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义的source读取MySQL中的数据

    val mysqlSource = new MysqlSource
    val mysqlDS: DataStream[String] = env.addSource(mysqlSource)

    mysqlDS.print()

    env.execute()


  }

  /**
   * 自定义source，实现sourceFunction接口
   */

  class MysqlSource extends SourceFunction[String]{
    /**
     *
     * run：用于读取外部数据的方法，只执行一次
     *
     * ctx：上下文对象，用于将读取到的数据发送到下游
     */
    override def run(sourceContext: SourceFunction.SourceContext[String]): Unit = {
      /**
       * 使用jdbc读取mysql数据，将读取到的数据发送到下游
       *
       */
      //加载驱动

      Class.forName("com.mysql.jdbc.Driver")

      //创建连接

      val conn: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/test", "root", "123456")

      //编写查询数据的sql
      val statement: PreparedStatement = conn.prepareStatement("select * from student")

      val resultSet: ResultSet = statement.executeQuery()

      //解析数据
      while (resultSet.next()){
        //通过字段名获取数据
        val id: Long = resultSet.getLong("id")
        val name: String = resultSet.getString("name")
        val age: Long = resultSet.getLong("age")
        val gender: String = resultSet.getString("gender")
        val clazz: String = resultSet.getString("clazz")

        //将每一条数据发送到下游
        sourceContext.collect(s"$id\t$name\t$age\t$gender\t$clazz")
      }
      //关闭连接
      statement.close()
      conn.close()

    }

    //任务被取消的时候执行，一般用于回收资源
    override def cancel(): Unit = {}
  }
}
